package com.bw.medical.realtime.dwd.app;

import com.bw.medical.realtime.common.base.BaseSQL;
import com.bw.medical.realtime.common.constant.Constant;
import com.bw.medical.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeConsultationPaySuc extends BaseSQL {
    public static void main(String[] args) {
        new DwdTradeConsultationPaySuc().start(
                10012,
                4,
                "medical_dwd_trade_consultation_pay_suc"
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
        readOdsDb(tabEnv,"dwd_trade_consultation_pay_suc");

        Table resultTable = tabEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['update_time'] consultation_pay_suc_time,\n" +
                "`data`['consultation_fee'] consultation_fee,\n" +
                "`data`['doctor_id'] doctor_id,\n" +
                "`data`['patient_id'] patient_id,\n" +
                "`data`['user_id'] user_id\n" +
                "from topic_db\n" +
                "where `table` = 'consultation'\n" +
                "and `type` = 'update'\n" +
                "and `data`['status'] = '203'");

        tabEnv.createTemporaryView("result_table",resultTable);

//        resultTable.execute().print();

        //创建动态表 和 要写入的主题进行映射
        tabEnv.executeSql("create table "+ Constant.DWD_TRADE_CONSULTATION_PAY_SUC+" (\n" +
                "`id` string,\n" +
                "`consultation_pay_suc_time` string,\n" +
                "`consultation_fee` string,\n" +
                "`doctor_id` string,\n" +
                "`patient_id` string,\n" +
                "`user_id` string\n" +
                ")" +  SQLUtil.getKafkaSinkSQL(Constant.DWD_TRADE_CONSULTATION_PAY_SUC)) ;

        resultTable.executeInsert(Constant.DWD_TRADE_CONSULTATION_PAY_SUC);
    }
}
